package com.xian.spark.mongo;

import com.mongodb.spark.MongoSpark;
import com.mongodb.spark.rdd.api.java.JavaMongoRDD;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.bson.Document;

/**
 * 获取mongo数据写入到hbase
 */
public class MongoToHbase {
    public static void main(String[] args) {
        //设置配置
        SparkConf sparkConf = new SparkConf();
        //sparkConf.setAppName("java wordcount local");
        //sparkConf.setMaster("local");
        //sparkConf.set("spark.mongodb.input.uri", "mongodb://dwro:uVtoSB8R5y45Sz_9@172.17.34.11:27017/driver.expDriverOnlineTimeDay?authSource=driver");
        //sparkConf.set("spark.mongodb.input.uri", "mongodb://172.18.5.150:20017/carorder.OrderExpCarInstant");
        sparkConf.set("spark.mongodb.input.uri", args[0]);

        JavaSparkContext jsc = new JavaSparkContext(sparkConf);

        //获取hbase数据
        JavaMongoRDD<Document> mongoRDD = MongoSpark.load(jsc);

        long count = mongoRDD.count();
        System.out.println("该表总记录数:"+count);


        JavaRDD<String> map = mongoRDD.map(new Function<Document, String>() {
            @Override
            public String call(Document document) throws Exception {
                return null;
            }
        });




        //关闭资源
        jsc.stop();
        jsc.close();
    }
}
